package org.dialMsg.Producer;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.dialMsg.Pojo.BatchPostParam;
import org.dialMsg.Pojo.TDialResult;
import org.dialMsg.Server.ServerImp.*;
import org.dialMsg.Server.ServerInt.RobotExecLogService;
import org.dialMsg.Topic.RocketmqConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@EnableAsync
public class SendTopicBatch1 {
    private static final Logger logger = LoggerFactory.getLogger(SendTopicBatch.class);
    private static final String BATCH_SIZE_KEY = "manage:creatAiData:batchSize";
    private static final String MAX_PROCESSING_CYCLES_KEY = "manage:creatAiData:maxProcessingCycles";
    private static final String BATCH_INTERVAL_KEY = "manage:creatAiData:batchInterval";
    private static final int DEFAULT_BATCH_SIZE = 500;
    private static final int DEFAULT_MAX_PROCESSING_CYCLES = 500;
    private static final int DEFAULT_BATCH_INTERVAL = 10; // ms
    private static final Random random = new Random();

    // 添加限流器：每秒80个请求
    private static final RateLimiter rateLimiter = RateLimiter.create(550.0);

    @Autowired
    private TDialResultImp tDialResultImp;

    @Autowired
    private PostExampleHutoolBatch client;

    @Autowired
    private FactMessageSendApiServerImp factMessageSendApiServerImp;

    @Autowired
    private RobotExecLogService robotExecLogService;

    // 线程安全的日期时间格式化器
    private static final DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    private static final String value = RocketmqConstant.MyTopic().get(RocketmqConstant.TopicEnum.ROBOT_DIAL_RESULT_MSG_TOPIC);
    private static final String phone = "111222333444";

    // 可配置的优化参数
    @Value("${processing.batch-size:100}")
    private int configBatchSize;

    @Value("${processing.max-cycles:100}")
    private int configMaxCycles;

    @Value("${processing.batch-interval:50}")
    private int configBatchInterval;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;


    public String sendMessageBatch() {
        Long lastMaxId = getLastMaxId();
        int batchSize = getBatchSize();
        int maxCycles = getMaxProcessingCycles();
        int batchInterval = getBatchInterval();
        AtomicInteger totalProcessed = new AtomicInteger(0);
        int cycleCount = 0;

        logger.info("开始批量处理数据. 最后处理ID: {}, 批大小: {}, 最大循环次数: {}",
                lastMaxId, batchSize, maxCycles);

        while (cycleCount < maxCycles) {
            cycleCount++;
            List<TDialResult> tDialResults = tDialResultImp.queryIdByPhoneAfter1(lastMaxId, phone, batchSize);

            if (tDialResults.isEmpty()) {
                logger.info("无新数据可处理，最后处理的ID: {}", lastMaxId);
                break;
            }

            AtomicLong currentMaxId = new AtomicLong(lastMaxId);
            List<Message<String>> messageList = new ArrayList<>(batchSize);
            List<String> allKeys = new ArrayList<>(batchSize);
            List<BatchPostParam> postParams = new ArrayList<>(batchSize);

            // 处理本批次数据
            tDialResults.forEach(record -> {
                processRecord(record, messageList, allKeys, postParams);
                currentMaxId.set(Math.max(currentMaxId.get(), record.getId()));
            });

            // 更新最后处理ID
            updateLastMaxId(currentMaxId.get());
            lastMaxId = currentMaxId.get();
            totalProcessed.addAndGet(tDialResults.size());

            // 异步处理本批次数据
            executeAsyncTasks(tDialResults, messageList, postParams);

            // 每批次处理后的日志
            logger.info("已处理第 {} 批数据, 处理数量: {}, 当前最大ID: {}",
                    cycleCount, tDialResults.size(), currentMaxId.get());

            // 当前批次不足batchSize说明是最后一批
            if (tDialResults.size() < batchSize) {
                logger.debug("处理到不足一批的数据, 结束循环");
                break;
            }

            // 批次间延迟（避免DB压力）
            sleepWithInterval(batchInterval, cycleCount);
        }

        // 全部处理完成后日志
        String resultMsg = totalProcessed.get() > 0 ?
                String.format("成功处理 %d 条数据, 批次: %d", totalProcessed.get(), cycleCount) :
                "没有可处理的数据";

        logger.info(resultMsg);
        return resultMsg;
    }

    //=== 私有方法 ===//

    /**
     * 从Redis获取最后处理的ID
     */
    private Long getLastMaxId() {
        Integer CurrentMaxId = tDialResultImp.queryCurrentMaxId();
        long l = Long.parseLong(String.valueOf(CurrentMaxId));
        try {
            String maxIdStr = redisTemplate.opsForValue().get("manage:creatAiData:lastMaxId");
            maxIdStr = CurrentMaxId >= Integer.parseInt(maxIdStr) ? String.valueOf(CurrentMaxId) : maxIdStr;
            return maxIdStr != null ? Long.parseLong(maxIdStr) : l;
        } catch (Exception e) {
            logger.warn("获取最后处理ID失败, 使用默认值0", e);
            return l;
        }
    }

    /**
     * 获取批处理大小
     */
    private int getBatchSize() {
        try {
            String batchSizeStr = redisTemplate.opsForValue().get(BATCH_SIZE_KEY);
            return batchSizeStr != null ? Integer.parseInt(batchSizeStr) :
                    configBatchSize > 0 ? configBatchSize : DEFAULT_BATCH_SIZE;
        } catch (Exception e) {
            logger.warn("获取批处理大小失败，使用默认值: {}", DEFAULT_BATCH_SIZE, e);
            return DEFAULT_BATCH_SIZE;
        }
    }

    /**
     * 获取最大处理批次数量
     */
    private int getMaxProcessingCycles() {
        try {
            String maxCyclesStr = redisTemplate.opsForValue().get(MAX_PROCESSING_CYCLES_KEY);
            return maxCyclesStr != null ? Integer.parseInt(maxCyclesStr) :
                    configMaxCycles > 0 ? configMaxCycles : DEFAULT_MAX_PROCESSING_CYCLES;
        } catch (Exception e) {
            logger.warn("获取最大处理批次失败，使用默认值: {}", DEFAULT_MAX_PROCESSING_CYCLES, e);
            return DEFAULT_MAX_PROCESSING_CYCLES;
        }
    }

    /**
     * 获取批次间隔时间
     */
    private int getBatchInterval() {
        try {
            String intervalStr = redisTemplate.opsForValue().get(BATCH_INTERVAL_KEY);
            return intervalStr != null ? Integer.parseInt(intervalStr) :
                    configBatchInterval > 0 ? configBatchInterval : DEFAULT_BATCH_INTERVAL;
        } catch (Exception e) {
            logger.warn("获取批次间隔时间失败，使用默认值: {}ms", DEFAULT_BATCH_INTERVAL, e);
            return DEFAULT_BATCH_INTERVAL;
        }
    }

    /**
     * 批次间休眠控制
     */
    private void sleepWithInterval(int batchInterval, int cycleCount) {
        if (batchInterval > 0 && cycleCount > 1) {
            try {
                Thread.sleep(batchInterval);
                logger.debug("批次间隔 {}ms", batchInterval);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.warn("批次间隔休眠被中断", e);
            }
        }
    }

    /**
     * 处理单个拨号记录
     */
    private void processRecord(TDialResult record,
                               List<Message<String>> messageList,
                               List<String> allKeys,
                               List<BatchPostParam> postParams) {
        // 生成消息唯一键
        String keys = dtf.format(LocalDateTime.now()) +
                RandomStringUtils.randomAlphanumeric(18).toUpperCase();
        allKeys.add(keys);
        // 构建MQ消息
        messageList.add(buildMqMessage(record, keys));

        // 构建POST参数
        int randomStatus = getRandomStatus();
        postParams.add(buildPostParam(record, randomStatus));
    }

    /**
     * 构建MQ消息
     */
    private Message<String> buildMqMessage(TDialResult record, String keys) {
        JSONObject dialData = new JSONObject();
        dialData.put("dialId", record.getId());
        dialData.put("callType", record.getCallType());
        dialData.put("batchId", record.getDataSourceBatchId());
        dialData.put("phone", record.getPhone());
        dialData.put("sceneId", record.getBusinessId());
        dialData.put("caller", record.getOutboundNumber());
        dialData.put("roundCount", record.getRoundCount());
        dialData.put("robotId", record.getRobotId());
        dialData.put("transferResult", record.getBridgeResult());
        dialData.put("durationOfSeatConversing", record.getDurationOfSeatConversing());
        dialData.put("seatConversingDuration", record.getDurationOfSeatConversing());
        dialData.put("dialStatus", record.getCallStatus());
        dialData.put("taskName", record.getShowName());
        dialData.put("batchName", record.getShowName());
        dialData.put("dataSourceType", record.getCallType());
        dialData.put("bankManager", record.getBankManager());
        dialData.put("bankOrgName", record.getBusinessName());
        dialData.put("batchDetailId", record.getDataSourceBatchDetailId());
        dialData.put("bankOrg", record.getBankOrg());
        dialData.put("followUserId", record.getFollowUser());

        JSONArray dataArray = new JSONArray();
        dataArray.add(dialData);
        JSONObject mainObject = new JSONObject();
        mainObject.put("accountId", record.getAccountId());
        mainObject.put("data", dataArray);
        String dialTaskMQMsg = JSON.toJSONString(mainObject);

        return MessageBuilder.withPayload(dialTaskMQMsg)
                .setHeader("KEYS", keys)
                .build();
    }


    /**
     * 更新Redis中的最大ID
     */
    private void updateLastMaxId(Long currentMaxId) {
        try {
            redisTemplate.opsForValue().set("manage:creatAiData:lastMaxId", currentMaxId.toString());
            logger.debug("更新最后处理ID成功: {}", currentMaxId);
        } catch (Exception e) {
            logger.error("更新最后处理ID失败", e);
        }
    }

    /**
     * 执行所有异步任务
     */
    private void executeAsyncTasks(List<TDialResult> records,
                                   List<Message<String>> messageList,
                                   List<BatchPostParam> postParams) {
        // 在任务提交前统一限流
        rateLimiter.acquire();
        // 提交异步任务
        sendMqMessagesAsync(messageList);
        sendPostRequestsAsync(postParams);
        syncReportDataAsync(records);

        // 生成语音日志（每条记录单独提交）
        for (TDialResult record : records) {
            generateSpeechLog(record);
        }
    }

    @Async("batchTaskExecutor")
    protected void generateSpeechLog(TDialResult record) {
        try {
            String content = "[{\"callContent\":\"喂（空1.2秒）您好，这里是珠海建行，您在我行有笔定期存款已经到期还未办理转存，我安排网点客户经理给您介绍下最新转存产品信息好吗？!(R130)\",\"callTime\":1739614767097,\"createTime\":\"2025-02-15 18:19:27\",\"flowNodeSeq\":\"R130\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"nodeId\":\"df831ec0-48dc-11ec-8ea2-2766b20a6b2d\",\"playbackUid\":\"bc68f93446cbd7d428ab52a4edec71bc\",\"robot\":true,\"skipStat\":false,\"tips\":\"开场白\",\"voiceCostTime\":0},{\"analyseType\":6,\"asrResult\":\"你说什么？我没听清。\",\"callContent\":\"你说什么？我没听清。(Q61)\",\"callContentCutPos\":\"你/r,说/v,什么/r,？/w,我/r,没/v,听/v,清/LOC,。/w\",\"callTime\":1739614785964,\"createTime\":\"2025-02-15 18:19:45\",\"flowNodeSeq\":\"Q61\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"qaEntity\":{\"highIntentionQa\":0,\"intentionTag\":\"What\",\"intentionType\":3,\"qaName\":\"227.tmpc_什么事\",\"qaUid\":61,\"qawords\":[{\"scroe\":0.5,\"word\":\"227.tmpc_什么事\"}],\"tagName\":\"227.tmpc_什么事\",\"targetWord\":\"你说#什么#\",\"type\":1},\"robot\":false,\"skipStat\":false,\"tips\":\"中性QA命中\",\"voiceCostTime\":0},{\"callContent\":\"是这样，来电是提醒您在我行有笔定期存款已经到期还未转存，建议您尽快办理转存，以免影响到您的收益。稍后我让客户经理给您具体介绍下转存产品信息及流程好吗？(R104)\",\"callTime\":1739614785964,\"createTime\":\"2025-02-15 18:19:45\",\"flowNodeSeq\":\"R104\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"playbackUid\":\"4997788b06bf0641fdd1d3a729e65e06\",\"robot\":true,\"skipStat\":false,\"voiceCostTime\":0},{\"analyseType\":7,\"asrResult\":\"你再说一遍，我没听清。\",\"callContent\":\"你再说一遍，我没听清。(Y4)\",\"callContentCutPos\":\"你/r,再说/d,一遍/m,，/w,我/r,没/v,听/v,清/LOC,。/w\",\"callTime\":1739614803510,\"createTime\":\"2025-02-15 18:20:03\",\"flowNodeSeq\":\"Y4\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"robot\":false,\"skipStat\":false,\"tips\":\"语态识别命中\",\"toneEntity\":{\"attribute\":6,\"attributeCategoryName\":\"sys.重复\",\"attributeCategoryUid\":4,\"targetWord\":\"Model\",\"tonewords\":[{\"scroe\":0.9999995231628418,\"value\":6,\"word\":\"sys.重复\"}]},\"voiceCostTime\":0},{\"callContent\":\"是这样，来电是提醒您在我行有笔定期存款已经到期还未转存，建议您尽快办理转存，以免影响到您的收益。稍后我让客户经理给您具体介绍下转存产品信息及流程好吗？(R104)\",\"callTime\":1739614803510,\"createTime\":\"2025-02-15 18:20:03\",\"flowNodeSeq\":\"R104\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"playbackUid\":\"4997788b06bf0641fdd1d3a729e65e06\",\"robot\":true,\"skipStat\":false,\"voiceCostTime\":0},{\"analyseType\":1,\"callContent\":\"被叫挂断(UE)\",\"callTime\":1739614814465,\"createTime\":\"2025-02-15 18:20:14\",\"flowNodeSeq\":\"UE\",\"hangup\":false,\"main\":false,\"matchCostTime\":0,\"needStudy\":false,\"robot\":false,\"skipStat\":false,\"tips\":\"被叫挂断\",\"voiceCostTime\":0,\"voiceText\":\"被叫挂断\"}]";
            robotExecLogService.addLog(record.getId(), content, record.getTimeCreate());
        } catch (Exception e) {
            logger.error("生成话单语音日志失败: 记录ID={}", record.getId(), e);
        }
    }

    @Async("batchTaskExecutor")
    protected void sendPostRequestsAsync(List<BatchPostParam> postParams) {
        if (postParams.isEmpty()) return;
        try {
            client.postBatchExample(postParams);
            logger.info("批量POST请求成功: {}条记录", postParams.size());
        } catch (Exception e) {
            logger.error("批量POST请求失败: {}条记录", postParams.size(), e);
        }
    }

    @Async("batchTaskExecutor")
    protected void sendMqMessagesAsync(List<Message<String>> messageList) {
        if (messageList.isEmpty()) return;

        try {
            // RocketMQ批量发送，每次最大100条
            int batchSize = 550;
            for (int i = 0; i < messageList.size(); i += batchSize) {
                int end = Math.min(i + batchSize, messageList.size());
                List<Message<String>> subList = messageList.subList(i, end);
                rocketMQTemplate.syncSend(value, subList, 5000);
                logger.info("已批量发送 {} 条MQ消息", subList.size());
            }
        } catch (Exception e) {
            logger.error("MQ消息批量发送失败", e);
        }
    }

    @Async("batchTaskExecutor")
    protected void syncReportDataAsync(List<TDialResult> records) {
        try {
            factMessageSendApiServerImp.sendFactMessage(records);
            logger.info("报表数据同步成功: {}条记录", records.size());
        } catch (Exception e) {
            logger.error("报表数据同步失败: {}条记录", records.size(), e);
        }
    }

    /**
     * 生成随机状态 (11, 20, 21)
     */
    private int getRandomStatus() {
        int[] statusArray = {11, 20, 21};
        return statusArray[random.nextInt(statusArray.length)];
    }

    /**
     * 构建POST参数对象
     */
    private BatchPostParam buildPostParam(TDialResult record, int randomStatus) {
        if (record.getCallType() == 1) {
            record.setDurationOfRinging(0);
            record.setDurationOfSeatConversing(0);
            record.setBridgeResult(0);
            record.setHumanStatus(10);
        }
        // 使用传统 SimpleDateFormat 格式化 Timestamp
//        java.text.SimpleDateFormat sdf = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//        String formattedTime = sdf.format(record.getTimeCreate());
        return new BatchPostParam(
                record.getBusinessId(),
                randomStatus,
                phone,
                record.getCallType(),
                record.getTimeCreate(),
                record.getBankManager(),
                record.getId(),
                record.getBridgeResult()
        );
    }
}